//package com.netty.esc.http.common.kafka;
//
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONObject;
//import com.netty.esc.http.common.es.EsUtils;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.kafka.support.Acknowledgment;
//import org.springframework.kafka.support.KafkaHeaders;
//import org.springframework.messaging.handler.annotation.Header;
//import org.springframework.scheduling.annotation.Async;
//import org.springframework.stereotype.Component;
//
//import javax.annotation.Resource;
//import java.io.IOException;
//import java.text.SimpleDateFormat;
//import java.util.Date;
//import java.util.Optional;
//
///**
// * @Author: zhaoyang05
// * @Data: 2022/3/24 17:29
// * @Description
// * 卡夫卡消费者
// */
//@Component
//@Slf4j
//public class Consumer {
//
//    @Resource
//    EsUtils esUtils;
//
//    private static volatile int NUM = 0;
//
//    private static final String TOPIC = "healthy";
//
//    private static final String TOPIC_GROUP1 = "FIRST_BLADE";
//
//    @Value("${fzxy.history}")
//    private Integer HISTORY;
//
//    @Value("${fzxy.history.history.name}")
//    private String HISTORY_NAME;
//
//    @Async
//    @KafkaListener(topics = TOPIC, groupId = TOPIC_GROUP1)
//    public void consumer(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) throws IOException {
//
//        Optional<?> message = Optional.ofNullable(record.value());
//        if (message.isPresent()) {
//            Object msg = message.get();
//            JSONObject jsonObject = JSONObject.parseObject(JSON.parse(message.get().toString()).toString());
//            System.out.println("状态为：：：" + NUM);
//            if (NUM==HISTORY)NUM=0;
//            if (NUM == 0){
//                jsonObject.put(HISTORY_NAME,1);
//            }else {
//                jsonObject.put(HISTORY_NAME,0);
//            }
//            NUM++;
//            jsonObject.put("createTime",new Date(jsonObject.getLong("createTime")));
//            log.info("topic_test 消费了： Topic:" + topic + ",Message:" + jsonObject);
//            Date date = new Date();
//            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
//            //每天生成一个新的index，防止数据过大
//            String humidity = jsonObject.getString("humidity");
//            String s = humidity.replaceAll("%", "");
//            jsonObject.put("humidity",s);
//            esUtils.add("fz" + sdf.format(date),"user",jsonObject);
//            ack.acknowledge();
//        }
//    }
//}
